/*
 * @Author: zhuzhenxxx
 * @Date: 2021-01-28 16:10:38
 * @LastEditTime: 2021-01-28 16:11:14
 * @LastEditors: Please set LastEditors
 * @Description: In User Settings Edit
 * @FilePath: /learn/heartserver.cpp
 */
#include <netinet/in.h>   // sockaddr_in
#include <sys/types.h>    // socket
#include <sys/socket.h>   // socket
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/select.h>   // select
#include <sys/ioctl.h>
#include <sys/time.h>
#include <iostream>
#include <vector>
#include <map>
#include <string.h>
#include <string>
#include <cstdlib>
#include <cstdio>
#include <cstring>
#include <mutex>
#include <thread>
#include <pthread.h>
#include "rtdbFun.h"
#include "fore_rtdb.h"
#include "rtdb_fun.h"
#include "./rapidjson/filereadstream.h"
#include "./rapidjson/document.h"
#include "./rapidjson/rapidjson.h" 
#include "./rapidjson/writer.h"
#include "./rapidjson/stringbuffer.h"
#include "./rapidjson/prettywriter.h"
#include "./rapidjson/filereadstream.h"
#include "./stdafx.h" 
#include "./my_mqtt.h"
#include "./mqtt_msg.h"
#include "./ZNet.h"
#include "./base.h"
#include "./_ccfree.h"
#include "_mysql.h"   // ccfree框架操作MySQL的头文件。
#include "tcpServer.h"


using namespace std;

#define BUFFER_SIZE 1024
typedef struct s_curve
{
	char dev_nodeID[64];
	char name[32];
	int  res0;
	char dev_type[32];
	char appName[32];
	char val[289*5];
	int  res1;
	int  index;
	char res2[64];
	char date[64];           //时间字符串
}s_curve;


int  readJson();
void Createdb();
int  qsort_yc_point( const void *a , const void *b );
void db_thread();
void printf_yc_rtdb( );

Server server(5000);

vector<long> vpthid; //存放线程id的容器

std::deque<s_package> vpackage;
const int DEQUESIZE = 1000;

pthread_mutex_t  dbmutex = PTHREAD_MUTEX_INITIALIZER;

pthread_mutex_t  upmutex = PTHREAD_MUTEX_INITIALIZER;

typedef struct s_table{
    char  id[128];
    char  pic_id[128];
    char  meta_id[128];
    char  item_id[128];
    char  prop_name[128];
    int   dev_id;
    char  data_id[64];
    char  data_name[64];
	char  update_time[64];
    int   update_tag;
}s_table;

std::deque<s_table> vtable;
pthread_mutex_t vtex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t vcnd = PTHREAD_COND_INITIALIZER;

std::map<std::string, std::string> mtable;   //tag  picid
std::map<std::string, std::string> mpro;    // tag  pro
std::map<std::string, std::string> mitem;    // tag  item

using namespace rapidjson;

void Createdb()
{
	EmptyAll(); 

	printf("CreateRtdb begin.....\n");
 
	CreateRTdb( "SYS_STRING_KEY_NAME",	SYS_STRING_KEY_NO, SYS_STRING_KEY_COUNT,sizeof(SYS_STRING_KEY));
	 
	CreateRTdb( "FORE_RTDB_YC_CURVE_5MIN",	FORE_CURVE_NO, FORE_CURVE_COUNT,sizeof(RTDB_YC_CURVE_5MIN)); 
	CreateRTdb( "FORE_YC_POINT_RTDB_NAME",	FORE_YC_POINT_NO, FORE_YC_POINT_COUNT,sizeof(FORE_YC_POINT)); 
	 	
	printf("CreateRtdb end.....\n");
}


int readJson()
{
	FILE* fp = fopen("config.json", "r"); // 非 Windows 平台使用 "r"
	 
	char readBuffer[65536];
	FileReadStream is(fp, readBuffer, sizeof(readBuffer));
	 
	Document root;
	root.ParseStream(is);

	if ( OpenDbms() == false ) {

		printf("open databases failed.\n");

	} else {

		printf("open databases success\n");
	}

	char  nodeId[64] = {0};
	char  meaName[32] = {0};
    //判断实时库是否被创建过
	if( !IsLibNoValid(FORE_YC_POINT_NO) || !IsLibNoValid(FORE_CURVE_NO) ) {

		Createdb();	
		SaveRtdb();		

		printf("Load rtdb rtdbms\n");
	} else {
		if( !LoadRtdb() ) {

			printf(" Load rtdb rtdbms failed2! \n"); 
		} else {

			printf(" Load rtdb rtdbms sucess2! \n"); 
		} 
	}

	// sleep(2);
	//插入记录（下面有savertdb 所以这里这调用一次就行，即第二次运行程序时，LoadRtdb函数会自动装载上次运行插入的记录，这里需要注意，通常这里插入记录就插入一次就行， ）

	std::string deviceType = root["deviceType"].GetString();

	std::string manufacturerId = root["manufacturerId"].GetString();

	const Value& services = root["services"];

	int ii=1;

	if (services.IsArray())
	{
		for (SizeType i = 0; i < services.Size(); i++) // 使用 SizeType 而不是 size_t
		{
			const Value& properties = services[i]["properties"];

			if( services[i]["properties"].IsArray())
			{
				for (SizeType j = 0; j < properties.Size(); j++)
				{
					std::string name = properties[j]["name"].GetString();

					std::cout << "name is "<< name << std::endl;
					RTDB_YC_CURVE_5MIN   yc_rec;

					sprintf(meaName, "%s", name.c_str() );  //对应xml中的  PhV_phsA、PhV_phsB等

					sprintf(nodeId, "%s:%s", manufacturerId.c_str(),  meaName );
					ii++;

					strcpy(yc_rec.dev_nodeID, nodeId);

					strcpy(yc_rec.name, meaName);

					UpdateRec_stringkey_Ex(FORE_CURVE_NO, nodeId, (char *)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));
				}
			}
		}
	}
		
	// conn.commit(); // 提交数据库事务。   
	std::cout << deviceType.c_str() << std::endl;
	std::cout << manufacturerId.c_str() << std::endl;

	fclose(fp);
	// printf_yc_rtdb();

	return 0;
}


int qsort_yc_point( const void *a , const void *b )
{
	RTDB_YC_CURVE_5MIN * p1 = (RTDB_YC_CURVE_5MIN*)a;
	RTDB_YC_CURVE_5MIN * p2 = (RTDB_YC_CURVE_5MIN*)b; 
 
	if( strcmp( p1->dev_nodeID, p2->dev_nodeID ) > 0 ) 
		return 1;

	else if( strcmp( p1->dev_nodeID, p2->dev_nodeID ) < 0 ) 
		return -1;

	else 
	{
		if( strcmp( p1->name, p2->name ) > 0 ) 
			return 1;

		else if( strcmp( p1->name, p2->name) < 0 ) 
			return -1;

		else return 0;
	}
}
 
void printf_5min_rtdb( )
{
	printf("\n\n -----------------printf_5minyc_rtdb--------- \n\n");

	tHandle handle = OpenLib(FORE_CURVE_NO);

	if( !IsHandleValid(handle) ) {

		printf("FORE_5minYC_POINT_NO is InValid \n");
		// return ;
	}
	
	std::vector<RTDB_YC_CURVE_5MIN>  aryYc; 

	RTDB_YC_CURVE_5MIN yc_rec;

	bool ret = ReadFirst(handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN ));

	while(ret)
	{  
		aryYc.push_back(yc_rec);  

		ret = ReadNext(handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN ));
	}
	
	CloseLib( handle );
	
	if( aryYc.size() > 0 )
		qsort(aryYc.data(), aryYc.size(), sizeof(RTDB_YC_CURVE_5MIN), qsort_yc_point );

	for( int i=0; i<aryYc.size(); i++ )
	{
		RTDB_YC_CURVE_5MIN& yc_rec = aryYc[i];

		printf("key = %lld, dev_nodeID = %s, name = %s CurVal0 = %.f CurVal1 = %.f CurVal2 = %.f index = %d val[229]=%.f val[230]=%.f val[231]=%.f\n", \
		yc_rec.KEY, yc_rec.dev_nodeID, yc_rec.name, yc_rec.CurVal0,yc_rec.CurVal1,yc_rec.CurVal2,yc_rec.index,yc_rec.val[229],yc_rec.val[230],yc_rec.val[231]);
		
		for(int ii = 0; ii < 289; ii++)
		{
			if(ii%10 == 0) printf("\n");
			printf("val[%3d]=%.f ", ii, yc_rec.val[ii]);
		}
		printf("\n\n\n\n");
	}
	 
	printf(" \n \n\n  count = %d  \n", aryYc.size() );
}


void printf_yc_rtdb( )
{
	printf("\n\n -----------------printf_yc_rtdb--------- \n\n");

	tHandle handle = OpenLib(FORE_YC_POINT_NO);

	if( !IsHandleValid(handle) ) {
		
		printf("FORE_YC_POINT_NO is InValid \n");
		return ;
	}
	
	std::vector<FORE_YC_POINT>  aryYc; 

	FORE_YC_POINT yc_rec;

	bool ret = ReadFirst(handle, (char*)&yc_rec, sizeof(FORE_YC_POINT ));

	while( ret )
	{  
			
		aryYc.push_back( yc_rec );  
		 
		ret = ReadNext(handle, (char*)&yc_rec, sizeof(FORE_YC_POINT ));
	}
	
	CloseLib( handle );
	
	if( aryYc.size() > 0 )
		qsort(aryYc.data(), aryYc.size(), sizeof(FORE_YC_POINT), qsort_yc_point );

	for( int i=0; i<aryYc.size(); i++ )
	{
		FORE_YC_POINT & yc_rec = aryYc[i];
		
		printf("key = %lld , dev_nodeID = %s , pro = %s , val = %0.3f , picid = %s itemid=%s \n", \
		 yc_rec.KEY, yc_rec.dev_nodeID, yc_rec.RES5, yc_rec.CurYcValue, yc_rec.picid, yc_rec.itemid);
	}
	 
	printf(" \n \n\n  count = %d  \n", aryYc.size() );
}


void db_thread()
{
	if ( OpenDbms() == false ) {
		printf("open databases failed.\n");
	} else {
		printf("open databases success\n");
	}

     //判断实时库是否被创建过
    if( !IsLibNoValid(SYS_STRING_KEY_NO) || !IsLibNoValid(FORE_YC_POINT_NO) ) {
         Createdb();
         
         SaveRtdb(FORE_YC_POINT_NO);
         
         printf("Load rtdb rtdbms\n");
	} else {
		if( !LoadRtdb() ) {

			printf(" Load rtdb rtdbms failed2! \n"); 

			return ;
		} else {

			printf(" Load rtdb rtdbms sucess2! \n"); 
		} 
	}

	char  name[32] = {0};
	char  nodeId[64] = {0};
	
	//队列没有值的时候不更新
	s_package dbre;

	tHandle							handle; 
	struct  FORE_YC_POINT		    curve_rec;

	while (true)
	{
		pthread_mutex_lock(&server.mapmutex);

		while(server.v_data.empty()) 
		{
			pthread_cond_wait(&server.cond,&server.mapmutex);
			// pthread_mutex_unlock(&server.mapmutex);
			sleep(1);
		}
	
		dbre = server.v_data.front();

		server.v_data.pop_front();
		
		pthread_mutex_unlock(&server.mapmutex);

		sprintf(nodeId, "%s", dbre.name);

		strcpy(curve_rec.dev_nodeID ,nodeId);

		FindRec(FORE_YC_POINT_NO, nodeId,(char*)&curve_rec, sizeof(FORE_YC_POINT));
       
		curve_rec.CurYcValue = dbre.val;
		
		auto tmp = mtable.find(curve_rec.dev_nodeID);
		auto tmp2 = mpro.find(curve_rec.dev_nodeID);
		auto tmp3 = mitem.find(curve_rec.dev_nodeID);

		if (STRCPY(curve_rec.picid, 128 , tmp->second.c_str())==0) continue;
		if (STRCPY(curve_rec.RES5, 32 , tmp2->second.c_str())==0) continue;
		if (STRCPY(curve_rec.itemid, 128 , tmp3->second.c_str()) == 0) continue;

		UpdateRec_stringkey_Ex(FORE_YC_POINT_NO, nodeId, (char *)&curve_rec, sizeof(FORE_YC_POINT));

		printf("yc tag name is %s data is %.lf \n", curve_rec.dev_nodeID, curve_rec.CurYcValue);

		SaveRtdb(FORE_YC_POINT_NO);
	}
}


void db_thread_5min()
{
	if ( OpenDbms() == false ) {

		printf("open databases failed.\n");
	} else {

		printf("open databases success\n");
	}
	if( !LoadRtdb() )  {

		printf(" Load rtdb rtdbms failed2! \n"); 
	} else {

		printf(" Load rtdb rtdbms sucess2! \n"); 
	} 
	
	char  nodeId[64] = {0};
	
	//队列没有值的时候不更新
	s_package dbre;
	RTDB_YC_CURVE_5MIN yc_rec;

	short 							curHour;
	short                           curDay;
	short							curMin;
	short							newIndex;
	struct  RTDB_YC_CURVE_5MIN		curve_rec;

	while (true)
	{
		struct tm   * curP;
	
		time_t tt = time(NULL); 

		curP = localtime(&tt);  

		curDay = curP->tm_mday; 
		curHour = curP->tm_hour;
		curMin = curP->tm_min;

		newIndex = curHour * 12 + curMin / 5 ;

		pthread_mutex_lock(&server.mapmutex);

		if(server.v_data.size() == 0 ) 
		{
			pthread_mutex_unlock(&server.mapmutex);
			sleep(2);
			continue;
		}
		else
		{
			dbre = server.v_data.front();
			// server.v_data.pop_front();
			pthread_mutex_unlock(&server.mapmutex);
		}

		sprintf(nodeId, "%s", dbre.name);

		strcpy(yc_rec.dev_nodeID ,nodeId);

		FindRec(FORE_CURVE_NO, nodeId,(char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));

		yc_rec.CurVal2 = yc_rec.CurVal1;      
		yc_rec.CurVal1 = yc_rec.CurVal0;           
		yc_rec.CurVal0 = dbre.val;

		yc_rec.val[newIndex] = dbre.val;

		auto tmp = mtable.find(yc_rec.dev_nodeID);
		auto tmp3 = mitem.find(yc_rec.dev_nodeID);

		if (STRCPY(yc_rec.res2, 64 , tmp->second.c_str())==0) continue;
		if (STRCPY(yc_rec.res3, 64 , tmp3->second.c_str()) == 0) continue;
		
		printf("yc_rec.val[newIndex]  is %.f\n",yc_rec.val[newIndex]);

		yc_rec.index = newIndex;
		UpdateRec_stringkey_Ex(FORE_CURVE_NO, nodeId, (char *)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));

		// 打印插入记录
		sleep(1); 
		// printf_5min_rtdb();
	}
}


void save_mysql()
{
	connection conn; // 数据库连接类
    s_curve curve;

	// 登录数据库，返回值：0-成功，其它-失败。
	// 失败代码在conn.m_cda.rc中，失败描述在conn.m_cda.message中。
	if (conn.connecttodb("127.0.0.1,root,lilian,jinan,3306","utf8") != 0)
	{
		printf("save_mysql connect database failed.\n%s\n",conn.m_cda.message);
	}

	if ( OpenDbms() == false )
	{
		printf("open databases failed.\n");
	}
	else
	{
		printf("open databases success\n");
	}
	if( !LoadRtdb() ) 
	{
		printf(" Load rtdb rtdbms failed2! \n"); 
	}
	else
	{
		printf(" Load rtdb rtdbms sucess2! \n"); 
	} 

	sqlstatement stmt2(&conn); // 操作SQL语句的对象。
	short 							curHour;
	short                           curDay;
	short							curMin;
	short							newIndex;

	time_t _lastlogtime;

	for ( ; ; )
	{
		sleep(5);	
		struct tm lt;
		struct tm nt;

		time_t now = time(0);

		memcpy(&lt, localtime(&_lastlogtime), sizeof(struct tm));
		memcpy(&nt, localtime(&now), sizeof(struct tm));

		if (lt.tm_yday == nt.tm_yday)
		{

		}
		else
		{
			_lastlogtime = now;
			
			tHandle							handle; 

			struct  RTDB_YC_CURVE_5MIN		curve_rec;

			handle = OpenLib(FORE_CURVE_NO);

			if( !IsHandleValid(handle) ) return ;

			RTDB_YC_CURVE_5MIN  yc_rec;			
			
			std::vector<RTDB_YC_CURVE_5MIN>  aryYc; 
	
			bool ret = ReadFirst( handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));

			while( ret )
			{
				
				if( FindRec(FORE_CURVE_NO, curve_rec.dev_nodeID,(char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN)) ) 
				{
					aryYc.push_back(yc_rec);  

					ret = ReadNext( handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));

					continue;
				}

				struct tm* curP;
	
				time_t tt = time(NULL); 
				curP = localtime(&tt);  

				curDay  = curP->tm_mday; 
				curHour = curP->tm_hour;
				curMin  = curP->tm_min;

				newIndex = curHour * 12 + curMin / 5 ;

				memset(&curve,0,sizeof(curve)); // 结构体变量初始化。

				// 为结构体变量的成员赋值。s
				sprintf(curve.dev_nodeID,"%s", yc_rec.dev_nodeID);     

				strcpy(curve.appName , yc_rec.appName);          

				strcpy(curve.name, yc_rec.name);

				curve.res0 = yc_rec.CurVal0;

				strcpy(curve.dev_type,yc_rec.dev_type);

				curve.res1 = yc_rec.CurVal1;    
    
				curve.index = yc_rec.index;

				for(int ii = 0; ii < 289; ii++)
				{
					char tmp[5];

					memset(tmp,0,sizeof(tmp));

					sprintf(tmp,"%.f,",yc_rec.val[ii]);

					strcat(curve.val,tmp);
				}

				printf("curve val is %s\n",curve.val);

				char strLocalTime[20];

				memset(strLocalTime,0,sizeof(strLocalTime));

				LocalTime(strLocalTime,"yyyy-mm-dd hh24:mi:ss");
				
				// 准备插入表的SQL语句。
				stmt2.prepare("insert into t_rttable(devnode,name,res0,dev_type,appname,res1,idx,res2,val) \
							values(:1,:2,:3,:4,:5,:6,:7,:8,:9)");

				// prepare方法不需要判断返回值。
				// 为SQL语句绑定输入变量的地址，bindin方法不需要判断返回值。
				stmt2.bindin(1, curve.dev_nodeID,64);
				stmt2.bindin(2, curve.name, 32);
				stmt2.bindin(3, &curve.res0);
				stmt2.bindin(4, curve.dev_type, 32);
				stmt2.bindin(5, curve.appName,32);
				stmt2.bindin(6, &curve.res1);
				stmt2.bindin(7, &curve.index);
				stmt2.bindin(8, curve.res2, 64);
				stmt2.bindin(9, curve.val, 1445);

				printf("strLocalTime is %s\n",strLocalTime);

				stmt2.bindin(10, strLocalTime, 19);

				// 执行SQL语句，一定要判断返回值，0-成功，其它-失败。
				// 失败代码在stmt.m_cda.rc中，失败描述在stmt.m_cda.message中。
				if (stmt2.execute() != 0)
				{
					printf("stmt.execute() failed.\n%s\n%s\n", stmt2.m_sql, stmt2.m_cda.message); 
				}
				else
				{
					printf("成功插入了%ld条记录。\n",stmt2.m_cda.rpc); // stmt.m_cda.rpc是本次执行SQL影响的记录数。
				}
				
				ret = ReadNext( handle, (char*)&yc_rec, sizeof(RTDB_YC_CURVE_5MIN));
			}

			CloseLib(handle);

			conn.commit(); // 提交数据库事务。    
			sleep(5);
			
			_lastlogtime = now;
		}
	}
}

CMyMQTTAsync	        g_myMqttClient;

int myRecv_Msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message); 

void exit_process( int signo )
{ 
	int pid = getpid();
	printf( "\n jsserver exit signo=%d pid=%d\n", signo, pid );
	
	exit(0);
}

int myRecv_Msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{ 
	MQTTAsync client = (MQTTAsync)context;
 
	char *  msg_buf = new char[message->payloadlen+1];
	memset( msg_buf, 0, message->payloadlen+1);
	memcpy(msg_buf, message->payload, message->payloadlen); 
 
	printf("Message arrived\n");
	printf("   topic: %s\n", topicName);
	printf("   message payloadlen=%d  : ", message->payloadlen);
  
	printf(" msg_buf = %s \n", msg_buf); 

	delete [] msg_buf;
	msg_buf = NULL; 

	MQTTAsync_freeMessage(&message);
	MQTTAsync_free(topicName);  

	return 1;
}

// string  toTimeString(time_t tt, bool IsBjTime )
// {
// 	//time_t lastTime = time(NULL) ; 
// 	struct tm *p = localtime(&tt);  

// 	int year  = p->tm_year+1900;
// 	int month = p->tm_mon+1;
// 	int day   = p->tm_mday; 
// 	int Hour   = p->tm_hour; 
// 	int Minute = p->tm_min;
// 	int second = p->tm_sec;

// 	char   chrToken[64] = {0} ;
// 	if( !IsBjTime ) {
// 		sprintf(chrToken, "%04d-%02d-%02dT%02d:%02d:%02dZ", p->tm_year+1900, p->tm_mon+1, p->tm_mday, p->tm_hour,p->tm_min, p->tm_sec);
// 	}
// 	else {
// 		sprintf(chrToken, "%04d-%02d-%02d %02d:%02d:%02d", p->tm_year+1900, p->tm_mon+1, p->tm_mday, p->tm_hour,p->tm_min, p->tm_sec);
// 	}

// 	return chrToken;
// }

bool _send_sub_data(CMyMQTTAsync * pCli,char* dev_type, char* nodeId ,float val)
{ 
	if( pCli == NULL || nodeId == NULL ) return false;

	int pos = 0;
	// int COUNT = rtData.size();
	int SEND_MAX_REC = 50;  
	int idx = 0;
	
	char  topic[128] = {0};
	sprintf(topic, "%s/terminal/dataReport/%s/%s", "dataCenterApp", \
			dev_type, nodeId);

	int  token = 10000;
 
	token = getpid();

	int updateTime = 0;

	rapidjson::StringBuffer s;
	rapidjson::Writer<rapidjson::StringBuffer> writer(s);
 	
	char  tmp[32] = {0}; 

	writer.StartObject();

	writer.Key("body");
	writer.StartObject();

	writer.Key("nodeId");
	sprintf(tmp, "%s", nodeId );
	writer.String(tmp);  

	writer.Key("val");
	sprintf(tmp, "%.2f", val );
	writer.String(tmp);  

	// writer.Key("serviceProperties"); 
	// writer.StartObject(); 

	// //dataProperties 
	// writer.Key("dataProperties"); 
	// writer.StartObject(); 

	writer.EndObject();

	writer.Key("token");
	sprintf(tmp, "%d", token );
	writer.String(tmp);
	writer.Key("timestamp"); 

	char tm[64];
	LocalTime(tm);
	writer.String( tm );

	writer.EndObject(); 

	printf( "send payload: %s\n", s.GetString());
	pCli->send_string(1, topic,  (char*)s.GetString() );
	 
	return true;
}

#include "./mqtt_cli.h"
using namespace TDbFun;

void mqtt_thread()
{
	char mqttSvrIp[32] = "tcp://127.0.0.1:1883";   
	char  mqtt_client_id[64] ={0};
	sprintf(mqtt_client_id, "testClientID_%d",  getpid() );

	while(1)
	{

		if( g_myMqttClient.connect_mqtt(mqtt_client_id, mqttSvrIp, myRecv_Msgarrvd) )
		{
			printf("connect_mqtt  ok !\n"); 
		}  
		else
		{
			continue;
		}
	 
		sleep(1);

		pthread_mutex_lock(&server.mapmutex);
		for (auto it = server.m_data.begin(); it != server.m_data.end(); it++)
		{
			char nodeId[64] ={0};
			snprintf(nodeId, "%s" , it->first.c_str(), 64);
			// _send_sub_data(&g_myMqttClient,devTypeInfo.deviceType ,nodeId,it->second);
		}
		pthread_mutex_unlock(&server.mapmutex);
	}
}

// bool _send_sub_data(void* context, int qos, FORE_DEV_DICT * pDevInfo, char* pServiceID, float val )
// { 
// 	if( context == NULL || pDevInfo == NULL ) return false;

// 	MQTTAsync  client = (MQTTAsync)context;
	
// 	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
// 	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	   
// 	int pos = 0;
// 	int SEND_MAX_REC = 50;  
// 	int idx = 0;

// 	char  topic[128] = {0};
// 	sprintf(topic, "%s/terminal/dataReport/%s/%s", "dataCenterApp", pDevInfo->deviceType, pDevInfo->dev_nodeID );

// 	int  token = 10000;
 
// 	token = getpid();

// 	rapidjson::StringBuffer s;
// 	rapidjson::Writer<rapidjson::StringBuffer> writer(s);
 	
// 	char  tmp[32] = {0}; 

// 	writer.StartObject();

// 	writer.Key("body");
// 	writer.StartObject();

// 	writer.Key("nodeId");
// 	sprintf(tmp, "%s", pDevInfo->dev_nodeID );
// 	writer.String(tmp);  

// 	writer.Key("val");
// 	sprintf(tmp, "%.2f", val);
// 	writer.String(tmp);  

// 	int  dataTime = 0;

// 	writer.EndObject();

// 	writer.Key("token");
// 	sprintf(tmp, "%d", token );
// 	writer.String(tmp);
// 	writer.Key("timestamp"); 

// 	// char tm[64];
// 	// LocalTime(tm);
// 	writer.String(base::toTimeString(dataTime).c_str());  
// 	// writer.String( tm );

// 	writer.EndObject(); 

// 	printf( "send payload: %s\n", s.GetString());
		 
// 	pubmsg.payload = (char*)s.GetString();
// 	pubmsg.payloadlen = s.GetSize() ;  
// 	pubmsg.retained = 0;
// 	pubmsg.qos = qos;
				
// 	opts.onSuccess =  onPubSend;
// 	opts.context = client;  

// 	base::my_printf( "send payload: %s\n", s.GetString());

//  	int rc = MQTTAsync_send(client, topic, pubmsg.payloadlen, pubmsg.payload,pubmsg.qos, pubmsg.retained, &opts);
// 	if ( rc  != MQTTASYNC_SUCCESS )
// 	{  
// 		base::my_printf( "Failed to  sendMessage  , return code %d\n", rc);
// 		return false;
// 	} 

// 	return true;
// }


void getmysql()
{
    connection conn2;

    if (conn2.connecttodb("127.0.0.1,root,lilian,jinan,3306","utf8") != 0)
	{
		printf("getmysql connect database failed.\n%s\n",conn2.m_cda.message);
	}

    sqlstatement stmt(&conn2); 
    s_table stable;

    stmt.prepare("SELECT id, pic_id, meta_id, item_id, prop_name, dev_id, data_id, data_name, date_format(update_time,'%%Y-%%m-%%d %%h:%%i:%%s') \
                    FROM `T_MEASURE` where update_tag=1");

    stmt.bindout(1, stable.id, 128);
    stmt.bindout(2, stable.pic_id,128);
    stmt.bindout(3, stable.meta_id,128);
    stmt.bindout(4, stable.item_id, 128);
    stmt.bindout(5, stable.prop_name, 128);
    stmt.bindout(6, &stable.dev_id);
    stmt.bindout(7, stable.data_id, 64);
    stmt.bindout(8, stable.data_name, 64);
    stmt.bindout(9, stable.update_time, 64);

    while(1)
    {
		vtable.clear();
        if(stmt.execute() != 0)
        {
            printf("stmt.excute() failedn.\n%s\n%s\n",stmt.m_sql,stmt.m_cda.message); 
        }

        while(1)
        {
            memset(&stable, 0, sizeof(stable));

            if(stmt.next() != 0) break;

			if(stable.pic_id==NULL || stable.prop_name == NULL || stable.item_id==NULL ) continue;

			pthread_mutex_lock(&vtex);
            vtable.push_back(stable);
			pthread_mutex_unlock(&vtex);
			pthread_cond_signal(&vcnd);

			mtable.insert(make_pair(stable.data_id, stable.pic_id));
			mpro.insert(make_pair(stable.data_id, stable.prop_name));
			mitem.insert(make_pair(stable.data_id, stable.item_id));

            printf("id=%s, pic_id=%s, meta_id=%s, item_id=%s, prop_name=%s, dev_id=%d, data_name=%s update_time=%s\n",\
                stable.id, stable.pic_id, stable.meta_id,stable.item_id,stable.prop_name,stable.dev_id,stable.data_name,stable.update_time);
        }

        printf("本次查询了table表%ld条记录。\n",stmt.m_cda.rpc);
        sleep(50);
    }
}

bool put_to_rtdb()
{
    if ( OpenDbms() == false )
	{
		printf("open databases failed.\n");
	}
	else
	{
		printf("open databases success\n");
	}

    //判断实时库是否被创建过
	if( !IsLibNoValid(SYS_STRING_KEY_NO) || !IsLibNoValid(FORE_YC_POINT_NO) ) {

		Createdb();	
		SaveRtdb();		
		printf("Load rtdb rtdbms\n");
	} else {
		if( !LoadRtdb() ) {
			printf(" Load rtdb rtdbms failed2! \n"); 
		} else {
			printf(" Load rtdb rtdbms sucess2! \n"); 
		} 
	}

	sleep(1);
    connection conn1;

    if (conn1.connecttodb("127.0.0.1,root,lilian,jinan,3306","utf8") != 0)
    {
        printf("put_to_rtdb connect database failed.\n%s\n",conn1.m_cda.message);
    }

    sqlstatement stmt1(&conn1); 

	char  nodeId[64] = {0};
	char  proName[64] = {0};
	char  item[64] = {0};
	char  pic[64] = {0};
	// char  
    FORE_YC_POINT   yc_rec;
    s_table dbre;

    while(1)
    {
		memset(nodeId, 0, sizeof(nodeId));
		memset(proName, 0, sizeof(proName));

        pthread_mutex_lock(&vtex);
		while(vtable.empty()) 
		{
			pthread_cond_wait(&vcnd, &vtex);
		}

		dbre = vtable.front();

		vtable.pop_front();

		pthread_mutex_unlock(&vtex);

        sprintf(proName, "%s", dbre.prop_name ); 
        sprintf(nodeId, "%s", dbre.data_id);
		sprintf(item, "%s", dbre.item_id);
		sprintf(pic, "%s", dbre.pic_id);

        strcpy(yc_rec.dev_nodeID, nodeId);
		if(yc_rec.dev_nodeID==NULL ) continue;

		auto tmp = mtable.find(yc_rec.dev_nodeID);
		auto tmp2 = mpro.find(yc_rec.dev_nodeID);
		auto tmp3 = mitem.find(yc_rec.dev_nodeID);

		if (STRCPY(yc_rec.picid, 128, tmp->second.c_str()) == 0) continue;
		if (STRCPY(yc_rec.RES5, 32, tmp2->second.c_str()) == 0) continue;
		if (STRCPY(yc_rec.itemid, 128, tmp3->second.c_str()) == 0) continue;

		// if ( (strcmp(yc_rec.picid, tmp->second.c_str()) == 0) && \
		// 	 (strcmp(yc_rec.RES5, tmp2->second.c_str()) == 0) && \
		// 	 (strcmp(yc_rec.itemid, tmp3->second.c_str()) == 0)  \
		// 	 ) 
		// {
		// 	continue;
		// }

        UpdateRec_stringkey(FORE_YC_POINT_NO, nodeId, (char *)&yc_rec, sizeof(FORE_YC_POINT));
        // printf_yc_rtdb();

        stmt1.prepare("update T_MEASURE set update_tag = :1 where dev_id = :2");

		int i=0;
		stmt1.bindin(1, &i);
        stmt1.bindin(2, &dbre.dev_id);
		printf("yc_rec.dev_nodeID is : %s\n", yc_rec.dev_nodeID);

        if(stmt1.execute() != 0)
        {
            printf("stmt1.excute() failedn.\n%s\n%s\n",stmt1.m_sql, stmt1.m_cda.message);
            return false;
        }

		conn1.commit();
    }
    return true;
}

void sysnDbThread()
{ 
    //启动获取数据线程
    //启动更新数据线程
    thread getdb   = thread(getmysql);
    thread putrtdb = thread(put_to_rtdb);

    getdb.join();
    putrtdb.join();
}

void start_tcp()
{
	// sleep(5);

	server.Bind();
    server.Listen();
    server.Run();
}

bool MY_PRINTF_TAG  = 0;
int main(int argc, char* argv[])
{
	signal(SIGKILL , exit_process);
	signal(SIGINT  , exit_process);
	signal(SIGTERM , exit_process);	 
	signal(SIGQUIT , exit_process);	 
	 
	if( argc == 2 ) { 
		if( strcmp(argv[1], "p") == 0 )
			MY_PRINTF_TAG = 1 ; 
	} 
	
	readJson();
	
	thread serv   = thread(start_tcp);
	thread syncdb = thread(sysnDbThread);
	// thread smysql = thread(save_mysql);
	thread dbthread      = thread(db_thread);
	// thread mqttthread = thread(mqtt_thread);
	thread dbthread_5min = thread(db_thread_5min);
	
	serv.join();
	// smysql.join();
	syncdb.join();
	dbthread.join();
	dbthread_5min.join();
	// mqttthread.join();

	return 0;
}
